package com.donglink.test.logmonitor.bolt;

import com.donglink.test.logmonitor.entry.App;
import com.donglink.test.logmonitor.entry.Message;
import com.donglink.test.logmonitor.service.EstimateService;
import com.donglink.test.logmonitor.service.impl.EstimateServiceImpl;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.List;
import java.util.Map;

public class EstimateBolt extends BaseRichBolt {
    OutputCollector outputCollector;
    EstimateService estimateService;
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector=outputCollector;
        this.estimateService =new EstimateServiceImpl();
    }

    @Override
    public void execute(Tuple tuple) {
        String log=tuple.getString(0);
        App app= (App) tuple.getValue(1);
        List<Message> messages=estimateService.proccess(log, app);
        if(messages.size()>0)
        {
            outputCollector.emit(new Values(messages));
        }
        else
        {
            return;
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("messages"));
    }
}
